
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
  
public class Step3 {  
    
public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {  
    private Text word = new Text();
	
    @Override 
    public void map(LongWritable key, Text value, Context context) throws IOException,  InterruptedException {  
        try {
    	if (!value.toString().isEmpty()){
        	String lineFromText = value.toString();
            String[] lineSplited = lineFromText.split("\t");
            String[] pairSplited = lineSplited[0].split(" ");
            String firstPair = pairSplited[0];
            String secondPair = pairSplited[1];
            String newPair = "";
            
            if (firstPair.compareTo(secondPair)<0)
            	 newPair = firstPair+" "+secondPair+" "+pairSplited[2];
            else
            	 newPair = secondPair+" "+firstPair+" "+pairSplited[2];
            
            word.set(newPair);
        	context.write(word, new Text(lineSplited[1]));
    	}
    	
		} catch (Exception e) {
			System.out.println("Error: In Step3");
		   }
      }
  }
  
  public static class ReduceClass extends Reducer<Text,Text,Text,Text> { 
	  public static final int M = 3;		
    @Override 
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException,  InterruptedException { 
    	List<String> valuesStr = new ArrayList<String>(); 
    	int countNumOfContexts = 0;
    	int sumNumerator = 0;
    	boolean isBiggerThanM = true;
    	
    	String[] keySplit = key.toString().split(" ");
    	for (Text value: values){
    		valuesStr.add(value.toString());
    		countNumOfContexts++;
    	}
    	
    	if (countNumOfContexts>=M){
    		for (String value : valuesStr) {
            	String[] valueStr = value.split("=");
            	String[] countNumOfAppearContext = valueStr[1].split(",");
            	int countOfContext1= Integer.parseInt(countNumOfAppearContext[0]);
            	int countOfContext2= Integer.parseInt(countNumOfAppearContext[1]);
            	int numerator= countOfContext1*countOfContext2;
            	sumNumerator += numerator;
            }
    	}
    	else{
    		isBiggerThanM = false;
    	}
    	if (isBiggerThanM){
    		double denominator= Double.parseDouble(keySplit[2]);
    		double similarity = sumNumerator / denominator;
    		context.write(new Text(keySplit[0]+" "+keySplit[1]), new Text(Double.toString(similarity)));
    	}
    }
  } 
  
 public static void main(String[] args) throws Exception {
	Path input = new Path(args[0]);
	Path output = new Path(args[1]);	
    Configuration conf = new Configuration();
    
    conf.set("mapred.max.split.size","5242880");
    
    Job job = new Job(conf, "Step3"); 
    job.setOutputKeyClass(Text.class); 
    job.setOutputValueClass(Text.class); 
    
    job.setJarByClass(Step3.class); 
    job.setMapperClass(MapClass.class); 
    job.setReducerClass(ReduceClass.class); 
    job.setInputFormatClass(TextInputFormat.class);
    job.setOutputFormatClass(TextOutputFormat.class);
    FileInputFormat.addInputPath(job, input); 
    FileOutputFormat.setOutputPath(job, output); 
    boolean flag = job.waitForCompletion(true); 
    if(!flag){
        System.err.println("Problem with Step3");
        return;
       }
  } 
}

  
